"""
案例分析:搜索引擎分析
"""

from pyspark import SparkContext,SparkConf
import os

os.environ["HADOOP_HOME"] = "D:/dev/spark-3.3.1-bin-hadoop3"
os.environ["PYSPARK_PYTHON"] = "C:/Users/86131/AppData/Local/Programs/Python/Python39/python.exe"
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
conf.set("spark.default.parallelism","1")
sc = SparkContext(conf=conf)
#读取文件创建RDD
rdd = sc.textFile("C:/Users/86131/Desktop/课程/python资料/资料/第15章资料/资料/search_log.txt")

#TODO 需求1: 热门搜索时间段Top3 (精确到小时)
#1.1 对字符串进行切割
#1.2 取出全部的时间转化为小时
#1.3 对每个时间段进行计数
#1.4 对热门事件进行排序取前三个
result1 = rdd.map(lambda x: (x.split("\t")[0][:2], 1)).\
    reduceByKey(lambda x,y: x+y).\
    sortBy(lambda x: x[1], ascending = False, numPartitions = 1).\
    take(3)
print(f"需求1的结果为: {result1}")

#TODO 需求2: 热门搜索词Top3
#2.1 对字符串进行切割
#2.2 取出搜索词
#2.3 对搜索词进行统计
#2.4 统计后排序
result2 = rdd.map(lambda x: (x.split("\t")[2], 1)).\
    reduceByKey(lambda x,y: x+y).\
    sortBy(lambda x: x[1], ascending = False, numPartitions = 1).\
    take(3)
print(f"需求2的结果为: {result2}")

#TODO 需求3: 统计黑马程序员关键字在什么时候被搜索的最多
#3.1 过滤内容,只保留黑马程序员的内容
#3.2 取出对应的搜索时间
#3.3 对时间进行统计
#3.4 对时间进行排序
result3 = rdd.map(lambda x: x.split("\t")).\
    filter(lambda x: x[2] =="黑马程序员").\
    map(lambda x: (x[0][:2],1)).\
    reduceByKey(lambda a,b: a+b).\
    sortBy(lambda x: x[1], ascending = False, numPartitions = 1).\
    take(1)

print(f"需求3的结果为: {result3}")


#TODO 需求4: 将数据转化为JSON格式, 写出到文件中
#4.1 转换为JSON格式RDD
#4.2 写出为文件
rdd.map(lambda x: x.split("\t")).\
    map(lambda x: {"time": x[0], "user_id": x[1], "key_word": x[2], "rank1": x[3], "rank2": x[4], "url": x[5]}).\
    saveAsTextFile("D:/output_json")













